package com.atguigu.loginfail_detect

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * @author: yangShen
  * @Description: 恶意登录监控： 如果一个用户短时间内频繁登录失败，就有可能是出现了程序的恶意攻击，比如密码暴力破解
  *              需求：在固定时间2秒内，如果登录失败达到2次则认为在恶意攻击
  *
  *              该实现存在的问题：1.数据乱序(eventTime，eventType)怎么办，第一次进来的时间大，第二次进来的时间小，举例相隔10秒，第二次 < 第一次 + 2 恒成立，就会造成只要错误两次就报警
  *                     2.指定的错误次数maxFailTimes没有用到
  * @Date: 2020/5/7 15:26 
  */

object LoginFailOptimize {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val source = getClass.getResource("/LoginLog.csv")
    val loginEventStream = environment.readTextFile(source.getPath)
      .map(data => {
        val dataArray = data.split(",")
        LoginEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
      })
      //设置时间戳
      //方式四：处理乱序数据，水位线waterMark设置为5毫秒，延时5毫秒上涨水位
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(5)) {
        override def extractTimestamp(element: LoginEvent): Long = element.eventTime * 1000L
      })

    val warningStream = loginEventStream
      .keyBy(_.userId)    //以用户ID进行分组
      //底层api，放大招
      .process(new LoginWarningOptimize(2))

    warningStream.print()

    environment.execute("login fail job")
  }
}

 class LoginWarningOptimize(maxFailTimes: Int) extends KeyedProcessFunction[Long, LoginEvent, Warning]{

  //定义状态，保存2秒内的所有登录失败事件
  lazy val loginFailState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("login-fail-state", classOf[LoginEvent]))

  override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#Context, out: Collector[Warning]): Unit = {
    if (value.eventType == "fail"){
      //如果是失败，判断之前之前是否有登录失败事件
      val iter = loginFailState.get().iterator()
      if (iter.hasNext){
        //如果已经有登录失败事件，则比较登录失败时间
        val firstFail = iter.next()
        //如果两次间隔小于2秒，输出报警
        if (value.eventTime < firstFail.eventTime  + 2){
          out.collect(Warning(value.userId, firstFail.eventTime, value.eventTime, "login fail in 2 seconds."))
        }
        //更新最近一次的登录失败事件，保存在状态里
        loginFailState.clear()
        loginFailState.add(value)
      }else {
        //如果是第一次登录失败，直接添加状态
        loginFailState.add(value)
      }
    }else {
      //如果是成功，清空状态
      loginFailState.clear()
    }
  }

  //定时器回调函数
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#OnTimerContext, out: Collector[Warning]): Unit = {

  }
}